package com.sne.kafka;

import com.alibaba.fastjson.JSONObject;
import com.sne.sys.entity.SysLog;
import com.sne.sys.entity.SysLoginLog;
import com.sne.sys.service.ISysLogService;
import com.sne.sys.service.ISysLoginService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @author fanghh
 * @date 2020/11/13
 */
@Slf4j
@Component
public class SysLoginConsumer {

    @Autowired
    private ISysLoginService iSysLoginService;

    @KafkaListener(topics = "sys-login")
    private void logConsumer(ConsumerRecord consumerRecord){
        try{
            Optional message = Optional.ofNullable(consumerRecord.value());
            if(message.isPresent()){
                SysLoginLog sysLogin = JSONObject.parseObject(message.get().toString(), SysLoginLog.class);
                log.info("消费{}",message);
                iSysLoginService.save(sysLogin);
            }
        }catch (Exception e){
            log.error("kafka消费异常",e);
        }
    }
}
